package org.apache.zookeeper.server.paxos2;

import java.net.ServerSocket;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.LinkedList;
import java.util.List;

import org.apache.zookeeper.proto.PrepareDataRequest;
import org.apache.zookeeper.server.PrepareDataCache;

/**
 * Created by IntelliJ IDEA. User: wangw Date: 2011-6-13 Time: 23:11:09 To
 * change this template use File | Settings | File Templates.
 */
public class SyncThread extends Thread {
	Storage storage;

	public void setStorage(Storage storage) {
		this.storage = storage;
	}
	
	
	/*public void run() {
		try {
			ServerSocket ss = new ServerSocket(5454);
			while (true) {
				Socket sock = ss.accept();
				ObjectInputStream ois = new ObjectInputStream(sock
						.getInputStream());
				ObjectOutputStream oos = new ObjectOutputStream(sock
						.getOutputStream());
				Object req = ois.readObject();
				if (req instanceof StartSync) {
					System.out.println("processSync start");
					StartSync startSync = (StartSync) req;
					List<Long> blocks = new LinkedList<Long>();
					List<Boolean> valid = new LinkedList<Boolean>();
					List<byte[]> blockMetas = new LinkedList<byte[]>();
					storage.getNewBlocks(startSync.zxid, blocks, valid,
							blockMetas);
					System.out.println("processSync assemble block array");
					int index = 0;
					long[] blocksArray = new long[blocks.size()];
					for (Long block : blocks) {
						blocksArray[index++] = block;
					}
					System.out.println("processSync assemble version array");
					index = 0;
					byte[][] blockMetasArray = new byte[blockMetas.size()][];
					for (byte[] blockMeta : blockMetas) {
						blockMetasArray[index++] = blockMeta;
					}
					boolean[] validArray = new boolean[valid.size()];
					index = 0;
					for (Boolean v : valid) {
						validArray[index++] = v;
					}

					StartSyncResponse resp = new StartSyncResponse();
					resp.blocks = blocksArray;
					resp.valid = validArray;
					resp.blockMetas = blockMetasArray;
					resp.zxid = startSync.zxid;
					System.out
							.println("processSync start sending StartSyncResponse");
					oos.writeObject(resp);
					oos.flush();
					sock.close();
					System.out
							.println("processSync end sending StartSyncResponse");
				}
				if (req instanceof FetchBlocks) {
					FetchBlocks fetchBlocks = (FetchBlocks) req;
					byte[][] blockMetas = fetchBlocks.blockMetas;
					byte[][] data = new byte[fetchBlocks.blockIds.length][];

					System.out.println("recover read "
							+ fetchBlocks.blockIds.length + " blocks");
					for (int i = 0; i < fetchBlocks.blockIds.length; i++) {
						BlockMeta meta = new BlockMeta(ByteBuffer
								.wrap(blockMetas[i]));
						PrepareDataRequest prepare = PrepareDataCache.get(
								meta.clientId, meta.cxid);
						if (prepare != null) {
							System.out
									.println("recover read from prepare cache");
							data[i] = prepare.getData();
						} else {
							data[i] = new byte[Configuration.blockSize];
							boolean valid = storage.readWithMeta(
									fetchBlocks.blockIds[i], data[i], meta);
							if (!valid)
								data[i] = null;
							meta.serialize(ByteBuffer.wrap(blockMetas[i]));
						}

					}
					FetchBlocksResponse resp = new FetchBlocksResponse();
					resp.blockIds = fetchBlocks.blockIds;
					resp.blockMetas = blockMetas;
					resp.data = data;
					oos.writeObject(resp);
					oos.close();
					sock.close();
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}*/
}
